package org.gruposp2p.cometd;

import com.infusion.util.event.EventBroker;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import javax.servlet.ServletContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.context.ServletContextAware;
import org.codehaus.groovy.grails.commons.ApplicationHolder;
import org.codehaus.groovy.grails.plugins.springsecurity.GrailsUser;
import org.cometd.bayeux.Message;
import org.cometd.bayeux.client.ClientSessionChannel;
import org.cometd.bayeux.server.BayeuxServer;
import org.cometd.bayeux.server.ConfigurableServerChannel;
import org.cometd.bayeux.server.ServerChannel;
import org.cometd.bayeux.server.ServerSession;
import org.cometd.oort.Oort;
import org.cometd.oort.Seti;
import org.cometd.server.AbstractService;
import org.cometd.server.ext.AcknowledgedMessagesExtension;
import org.cometd.server.ext.TimesyncExtension;
import org.cometd.server.filter.DataFilterMessageListener;
import org.cometd.server.filter.JSONDataFilter;
import org.cometd.server.filter.NoMarkupFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import grails.plugins.springsecurity.SpringSecurityService;

/*
 * Para inicializar cluster descomentar las líneas Seti y Oort en web.xml
 */
public class KarmaMessageService extends AbstractService implements ServletContextAware,
    ClientSessionChannel.MessageListener, BayeuxServer.ChannelListener, BayeuxServer.SubscriptionListener{

    private static Logger logger = LoggerFactory.getLogger(KarmaMessageService.class);
	
    private final ConcurrentMap<String, HashSet<String>> peers =
		new ConcurrentHashMap<String, HashSet<String>>();
    private final ConcurrentMap<String, String> peersSessions =
		new ConcurrentHashMap<String, String>();
    private Oort oort;
    private Seti seti;
    private ServletContext servletContext;
    
    //@Autowired EventBroker eventBroker;
    AlertChannelMessageProcessor alertMessageProcessor;
    SpringSecurityService springSecurityService;

    @Autowired
    public KarmaMessageService(BayeuxServer bayeux) {
        super(bayeux, "group");
        ApplicationContext context = (ApplicationContext)ApplicationHolder.getApplication().getMainContext();
        springSecurityService = (SpringSecurityService) context.getBean("springSecurityService");
        alertMessageProcessor = new AlertChannelMessageProcessor (this);
        bayeux.addExtension(new TimesyncExtension());
        bayeux.addExtension(new AcknowledgedMessagesExtension());
        bayeux.addListener(this);
    }
    
    @Override
    public void setServletContext(ServletContext servletContext) {
        logger.debug("***** setting Servlet Context *****");
        this.servletContext = servletContext;
        initCluster();
    }
	
    public void handleAreaMessage(ServerSession remoteClient, Message message) {
        logger.debug(" handleAreaMessage - message: " + message);
    }
    
    public void handleMetaMessage(ServerSession remoteClient, Message message) {
    	logger.debug(" handleMetaMessage - message: " + message);
        //eventBroker.publish(KarmaMessageEvent.NAME, new KarmaMessageEvent("campmund"));
    }
    
    public void handleAlert(ServerSession remoteClient, Message message) {
        logger.debug("springSecurityService.getPrincipal(): " + 
        		springSecurityService.getPrincipal().toString());
        alertMessageProcessor.processClientMessage(remoteClient, message);
    }

    public String sendMessageToPeer (
            String message, String channel, String peerId) {
        logger.debug(" - message: " + message + " - channel: " + channel + " - peerId: " + peerId);
        Set<String> channelMembers = peers.get(channel);
        if (channelMembers == null) return "Channel '" + channel + "' doesn't exist";
        if (!channelMembers.contains(peerId)) return "Peerid: " + peerId + " not found";
        seti.sendMessage(peerId, channel, message);
        return "OK";
    }

    private void initCluster () {
        /*Enumeration attNames = servletContext.getAttributeNames();
        while (attNames.hasMoreElements()) {logger.debug("attribute: " + attNames.nextElement());}*/
        //TODO
        Runnable runnable = new Runnable () {
            public void run() {
                int i = 0;
                while (oort == null || seti == null) {
                    logger.debug(" @@@@@@ CLUSTER INITIALIZATION - ITERATION: " + i++);
                    oort = (Oort)servletContext.getAttribute(Oort.OORT_ATTRIBUTE);
                    seti = (Seti)servletContext.getAttribute(Seti.SETI_ATTRIBUTE);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException ex) {
                         logger.error(ex.getMessage(), ex);
                    }
                }
                final DataFilterMessageListener noMarkup = new DataFilterMessageListener(
                getBayeux(),new NoMarkupFilter(),new KarmaMessageFilter());
                ServerChannel.Initializer initNoMarkup = new ServerChannel.Initializer() {
                    @Override
                    public void configureChannel(ConfigurableServerChannel channel) {
                        channel.setPersistent(true);
                        channel.addListener(noMarkup);
                    }
                };
                if (!getBayeux().createIfAbsent("/service/area/**",initNoMarkup) ||
                    !getBayeux().createIfAbsent("/alerts",initNoMarkup))
                    throw new IllegalStateException();
                addService("/**", "handleMetaMessage");
                addService("/service/area/**", "handleAreaMessage");
                addService(AlertChannelMessageProcessor.CHANNEL_NAME, "handleAlert");
                oort.observeChannel("/alerts");
                logger.debug(" @@@@@@ CLUSTER INITIALIZATED @@@@ ");
            }
        };
        new Thread(runnable).start();
    }

    /**
     * @return the oort
     */
    public Oort getOort() {
        return oort;
    }

    /**
     * @return the seti
     */
    public Seti getSeti() {
        return seti;
    }

   public void onMessage(ClientSessionChannel csc, Message msg) {
        logger.debug(" onMessage - ClientSessionChannel: " + csc + " - message: " + msg); 
    }

    public void channelAdded(ServerChannel sc) {
        logger.debug(" channelAdded - ServerChannel: " + sc + " - sc.getId(): " + sc.getId());
        peers.put(sc.getId(), new HashSet<String>());
    }

    public void channelRemoved(String string) {
        logger.debug(" channelRemoved - " + string);
        peers.remove(string);
    }

    public void configureChannel(ConfigurableServerChannel csc) {
        logger.debug(" configureChannel - ConfigurableServerChannel: " + csc);
    }

    public void subscribed(ServerSession serverSession, ServerChannel serverChannel) {
    	GrailsUser peer = (GrailsUser)springSecurityService.getPrincipal();
        logger.debug(" subscribed - ServerSession: " + serverSession +
                " - ServerChannel.getId(): " + serverChannel.getId() + 
                " - Peer id: " + ((peer == null)? "null" : peer.getId()));
        if (peer != null) {
        	Set<String> channelPeers = peers.get(serverChannel.getId());
        	seti.associate(String.valueOf(peer.getId()), serverSession);
        	channelPeers.add(String.valueOf(peer.getId()));
        	peersSessions.put(serverSession.getId(), String.valueOf(peer.getId()));
        }
    }

    public void unsubscribed(ServerSession serverSession, ServerChannel serverChannel) {
    	logger.debug(" unsubscribed - ServerSession: " + serverSession +
                " - ServerChannel: " + serverChannel + 
                " - Peer id: " + peersSessions.get(serverSession.getId()));
    	seti.disassociate(peersSessions.get(serverSession.getId())); 
    	String peerId = peersSessions.remove(serverSession.getId());
    	Set<String> channelMembers = peers.get(serverChannel.getId());
    	channelMembers.remove(peerId);
    }

    class KarmaMessageFilter extends JSONDataFilter {
        @Override
        protected Object filterString(String string) {
            logger.debug(" KarmaMessageFilter.filterString(): " + string);
            return string;
        }
    }

}